#include <stdio.h>
#include <stdlib.h>     //exit()
#include <string.h>
#include <errno.h>
#include <fcntl.h>      // 非阻塞宏
#include <sys/ioctl.h>
#include <sys/epoll.h>  // epoll管理服务器的连接和接收触发
#include <sys/socket.h>
#include <netinet/ip.h>
#include <pthread.h>    // 使用多线程
#include "websocket_common.h"
#include "websocket_server.h"
#include <sys/shm.h>  // shared memory
#include <sys/sem.h>  // semaphore
#include <sys/msg.h>  // message queue

extern int webSocket_clientLinkToServer(char *ip, int port, char *interface_path);
extern int webSocket_serverLinkToClient(int fd, char *recvBuf, int bufLen);
extern int webSocket_recv(int fd, DynBuf *pdata,  WebsocketData_Type *dataType);
typedef struct{
    int fd;
    int client_fd_array[EPOLL_RESPOND_NUM][2];//客户句柄数据
    char ip[128];
    int port;    
    CallBackFun callBack;
    CallCloseedFun callCloseed;
}WebSocket_Server;


/*************************
 * 线程 参数
 * ***********************/
typedef struct{
    int fd;
    WebsocketData_Type datatype;
    char *buf;
    int bufLen;
    CallBackFun callBack;
    CallCloseedFun callCloseed;
}pthreadArg;
/***************************
 * 回调线程
 * ********************/
void* thread_fun(void *arge)
{
    pthread_detach(pthread_self());

    pthreadArg *pArg=(pthreadArg*)arge;
    if(pArg->callBack!=NULL)
        pArg->callBack(pArg->fd,pArg->datatype,pArg->buf,pArg->bufLen);
    //printf("消息：%s\r\n",pArg->buf);
    if(pArg->buf){
         free(pArg->buf);
         //printf("放内存%d\r\n",pArg->buf);
    }
    if(pArg) free(pArg);
    //pthread_exit(0);

    return 0;
}
/*************************
 * 关闭事件线程
 * *********************/
void* thread_closeedfun(void *arge)
{
    pthread_detach(pthread_self());

    pthreadArg *pArg=(pthreadArg*)arge;
    if(pArg->callCloseed!=NULL)
        pArg->callCloseed(pArg->fd);
    //printf("消息：%s\r\n",pArg->buf);
    if(pArg->buf) free(pArg->buf);
    if(pArg) free(pArg);
   // pthread_exit(0);

    return 0;
}
/*****************************************************************
 * 添加值到数据组中
 * 参数：array[][2]数组
 *      arraySize数组大小
 *      value:值
 * 返回：0 成功，-1失败
*****************************************************************/
int arrayAddItem(int array[][2], int arraySize, int value)
{
    int i;
    for(i = 0; i < arraySize; i++)
    {
        if(array[i][1] == 0)
        {
            array[i][0] = value;
            array[i][1] = 1;
            return 0;
        }
    }
    return -1;
}
/*************************************************************
 * 从数组中移除值
 * 返回值： >=0 成功 序号 -1失败
***********************************************************/
int arrayRemoveItem(int array[][2], int arraySize, int value)
{
    int i;
    for(i = 0; i < arraySize; i++)
    {
        if(array[i][0] == value)
        {
            array[i][0] = 0;
            array[i][1] = 0;
            return i;
        }
    }
    return -1;
}
/************************************************************
 * 再找项
************************************************************/
int arrayFindItem(int array[][2], int arraySize, int value)
{
    int i;
    for(i = 0; i < arraySize; i++)
    {
        if(array[i][0] == value)
        {
            return i;
        }
    }
    return -1;
}
/*********************************
 * 再找项
 * 
 * **********************************/
int arrayFindNickname(char *array[], int arraySize, char *value)
{
    int i;
    for(i = 0; i < arraySize; i++)
    {
        if(strlen(array[i])>0)
        if((strstr(array[i],value)!=NULL)&&(strstr(array[i],value)==array[i]))
        {
            return i;
        }
    }
    return -1;
}
/*******************************
 * IPC全局变量
 * *****************************/
static char *shm=NULL;
static int shmid, semid, msqid;   
/***********************************
 * IPC通讯线程
 * *********************************/
void* thread_ipc_fun(void *arge)
{    
    struct msg_form msg;
    epollFD_info *pArg=(epollFD_info*)arge;
    //向共享内存写入数据
            if((shm!=NULL)&&(pArg!=NULL)){
                pthread_testcancel();//主动设置取消点
                    sem_p(semid);  /*访问资源*/                
                    memcpy(shm,pArg,FIFO_SIZE);
                    sem_v(semid);  /*释放资源*/                
                    msg.mtype = 9753;
                    msg.mtext = 'r';  /*发送消息通知读数据*/
                    msgsnd(msqid, &msg, sizeof(msg.mtext), 0);                
            }
    pthread_exit((void*)0);
}
/******************************************
*套接字监控进程
*****************************************/
void* server_thread_fun(void *arge)
{
	int ret , i , j;
    int removeRet;
	int accept_fd;
	int socAddrLen;
	struct sockaddr_in acceptAddr;
	struct sockaddr_in serverAddr;
    DynBuf buf;
    webSocket_bufInit(&buf,RECV_BYTE_LEN);
	//
	WebSocket_Server *wss = (WebSocket_Server *)arge;
	//
	memset(&serverAddr , 0 , sizeof(serverAddr)); 	// 数据初始化--清零     
    serverAddr.sin_family = AF_INET; 				// 设置为IP通信     
    //serverAddr.sin_addr.s_addr = inet_addr(wss->ip);// 服务器IP地址
    serverAddr.sin_addr.s_addr = INADDR_ANY;		// 服务器IP地址
    serverAddr.sin_port = htons(wss->port); 		// 服务器端口号    
	//
	socAddrLen = sizeof(struct sockaddr_in);
	
	//------------------------------------------------------------------------------ socket init
	//socket init
	wss->fd = socket(AF_INET, SOCK_STREAM,0);  
    if(wss->fd <= 0)  
    {  
        printf("创建套接字失败 !\r\n"); 
        exit(1);
    }    
    
    //设置为非阻塞接收
    ret = fcntl(wss->fd , F_GETFL , 0);
    fcntl(wss->fd , F_SETFL , ret | O_NONBLOCK);
    
    //bind sockfd & addr  
    while(bind(wss->fd, (struct sockaddr *)&serverAddr, sizeof(struct sockaddr)) < 0 )
        webSocket_delayms(1);
    
    //listen sockfd   
    ret = listen(wss->fd, 0);  
    if(ret < 0) 
    {  
        printf("server cannot listen request\r\n"); 
        close(wss->fd); 
        exit(1);
    } 

    
    //------------------------------------------------------------------------------ epoll init
     // 创建一个epoll句柄  
    int epoll_fd;  
    epoll_fd = epoll_create(EPOLL_RESPOND_NUM);  
    if(epoll_fd < 0)
    {  
        printf("server epoll_create failed\r\n"); 
        exit(1);
    }  
    
    int nfds;                     // epoll监听事件发生的个数  
    struct epoll_event ev;        // epoll事件结构体  
    struct epoll_event events[EPOLL_RESPOND_NUM];
    ev.events = EPOLLIN|EPOLLET;  // 	EPOLLIN		EPOLLET;    监听事件类型
    ev.data.fd = wss->fd;  
    // 向epoll注册server_sockfd监听事件  
    if(epoll_ctl(epoll_fd , EPOLL_CTL_ADD , wss->fd , &ev) < 0)
    {  
        printf("server epll_ctl : wss->fd register failed\r\n"); 
        close(epoll_fd);
        exit(1);
    }  

    //---------------------------------------共享内存------------------------
    key_t key;     
    struct shmid_ds buf1;  /*用于删除共享内存*/
    struct msqid_ds buf2;  /*用于删除消息队列*/
    
    epollFD_info fifo_buf;
    

    // 获取key值
    if((key = ftok(".", 'z')) < 0)
    {
        perror("ftok error");
        goto WSSTART;
    }

    // 创建共享内存
    if((shmid = shmget(key, FIFO_SIZE, IPC_CREAT|0666)) == -1)
    {
        perror("Create Shared Memory Error");
        goto WSSTART;
    }

    // 连接共享内存
    shm = (char*)shmat(shmid, 0, 0);
    if((int)shm == -1)
    {
        perror("Attach Shared Memory Error");
        goto WSSTART;
    }

    // 创建消息队列
    if ((msqid = msgget(key, IPC_CREAT|0777)) == -1)
    {
        perror("msgget error");
        goto WSSTART;
    }
    // 创建信号量
    semid = creat_sem(key);
    pthread_t thread_ipc;
	//------------------------------------------------------------------------------ server receive
WSSTART:    
    printf("\r\n\r\n==========WebSocket server start ! ==========\r\n\r\n",epoll_fd);

    while(1)
    { 
    	// 等待事件发生  
        nfds = epoll_wait(epoll_fd , events , EPOLL_RESPOND_NUM , -1);  	// -1表示阻塞、其它数值为超时
        if(nfds < 0)
        {  
            printf("server start epoll_wait failed\r\n"); 
            if(shm!=NULL)   shmdt(shm);
            close(epoll_fd);
            exit(1);
        }  
      //printf("工作线程:%d\n",pthreadNumber); 
        for(j = 0 ; j < nfds ; j++)
        { 
            fifo_buf.index=j;
            fifo_buf.fd=events[j].data.fd;
            memset(fifo_buf.status,0,16);
            fifo_buf.bufLen=0;
            fifo_buf.threadID=0;
        	//===================epoll错误 =================== 
            if ((events[j].events & EPOLLERR) || (events[j].events & EPOLLHUP))  
            {
            	    //printf("server epoll err !\r\n"); 
            	    //exit(1);
                    printf("epoll错误 accept close : %d\r\n", events[j].data.fd);     // 与客户端连接出错, 主动断开当前 连接
					// 向epoll删除client_sockfd监听事件  
					//ev.events = EPOLLIN|EPOLLET;
        			ev.data.fd = events[j].data.fd; 
				    if(epoll_ctl(epoll_fd , EPOLL_CTL_DEL , events[j].data.fd , &ev) < 0) 
				    {  
				        printf("server epoll_ctl : EPOLL_CTL_DEL failed !\r\n"); 
                        if(shm!=NULL)        shmdt(shm);
				        close(epoll_fd);
				        exit(1);
				    }
				    removeRet=arrayRemoveItem(wss->client_fd_array, EPOLL_RESPOND_NUM, events[j].data.fd);    // 从数组剔除fd
                    int cfd= events[j].data.fd;
                    close(events[j].data.fd);	//关闭通道
                    sprintf(fifo_buf.status,"closeed");
                        if(wss->callCloseed!=NULL)
                        {
                        //wss->callCloseed(cfd);
                            pthreadArg *pArg=(pthreadArg*)malloc(sizeof(pthreadArg));
                            memset(pArg,0,sizeof(pthreadArg));
                            pArg->fd=cfd;
                            pArg->datatype=WDT_NULL;
                            pArg->buf=NULL;
                            pArg->bufLen=0;
                            pArg->callBack=NULL;
                            pArg->callCloseed=wss->callCloseed;
                            pthread_t thread_id;
                            if(pthread_create(&thread_id, NULL, thread_closeedfun, (void *)(pArg)) != 0)
	                        {
		                    printf("创建closeed处理线程失败!\r\n");
                                if(pArg) free(pArg);
		                    continue;
                                }

                        }


            }
        	//===================新通道接入事件===================
        	else if(events[j].data.fd == wss->fd)
        	{		
				//轮巡可能接入的新通道 并把通道号记录在accept_fd[]数组中
				accept_fd = accept(wss->fd, (struct sockaddr *)&acceptAddr, &socAddrLen);                  
				if(accept_fd >= 0)  //----------有新接入，通道号加1----------
				{ 	
					// 向epoll注册client_sockfd监听事件  
					//ev.events = EPOLLIN|EPOLLET;
		    		ev.data.fd = accept_fd; 
					if(epoll_ctl(epoll_fd , EPOLL_CTL_ADD , accept_fd , &ev) < 0) 
					{  
						 printf("server epoll_ctl : EPOLL_CTL_ADD failed !\r\n"); 
                         if(shm!=NULL)        shmdt(shm);
						 close(epoll_fd);
						 exit(1);
					}
					//send(accept_fd , "OK\r\n" , 4 , MSG_NOSIGNAL);
                                        //printf("server fd/%d : accept\r\n", accept_fd);
					arrayAddItem(wss->client_fd_array, EPOLL_RESPOND_NUM, accept_fd);    // 添加fd到数组
                    sprintf(fifo_buf.status,"accept");
				}
			}
			//===================接收数据事件===================
			else if(events[j].events & EPOLLIN)
			{   
                webSocket_bufReset(&buf);            	
				//memset(buf, 0, sizeof(buf));
				
                WebsocketData_Type datatype;
                ret = webSocket_recv(events[j].data.fd , &buf ,  &datatype);    // 使用websocket recv
                if(ret > 0)
	            {
                    /*
		            printf("server(fd/%d) recv:\r\n"
                        "数据类型：%d\r\n"
                        "数据长度byte:%d\r\n"
                        "内容：\r\n%s\r\n"
                        , events[j].data.fd,datatype,ret, dataBuf);	*/	
                    if(wss->callBack!=NULL){
                        //wss->callBack(events[j].data.fd,datatype,dataBuf,ret);
                        char *pthreadBuf=(char*)calloc(ret,sizeof(char));//缓存创建后，放入了线程中。由线程完成用后处理
                        memcpy(pthreadBuf,buf.data,ret);
                        pthreadArg *pArg=(pthreadArg*)malloc(sizeof(pthreadArg));
                        memset(pArg,0,sizeof(pthreadArg));
                        pArg->fd=events[j].data.fd;
                        pArg->datatype=datatype;
                        pArg->buf=pthreadBuf;
                        pArg->bufLen=ret;
                        fifo_buf.bufLen=pArg->bufLen;
                        pArg->callBack=wss->callBack;
                        pArg->callCloseed=NULL;
                        pthread_t thread_id;
                        int ret_thread=pthread_create(&thread_id, NULL, thread_fun, (void *)(pArg));
                        if(ret_thread != 0)
	                    {
                                    printf("创建callBack处理线程失败,errno:%d!\r\n",ret_thread);
                                    /**************************
                                     * 通过反复的 pthread_create() --->   pthread_exit (0)   一段时间后，会导致pthread_create() 失败，返回11
                                     * 1. 主线程这里创建完子线程后，把子线程分离出来，线程运行结束后，让系统自动回收资源
                                     *   pthread_detach(thread_id)（非阻塞，可立即返回）
                                     *   2. 或者在线程函数开头，分离自己
                                     *       pthread_detach(pthread_self())
                                     *
                                     * ************************************/
                            if(pthreadBuf) free(pthreadBuf);
                            if(pArg) free(pArg);
		                    continue;
	                    } 
                        fifo_buf.threadID=thread_id;
                        sprintf(fifo_buf.status,"forward");


                    }
	            }
                else if(ret < 0)
                {
                    
                    if(strncmp(buf.data, "GET", 3) == 0)	//握手,建立连接
                    {
                            ret = webSocket_serverLinkToClient(events[j].data.fd, buf.data, ret);
                    }
                }
                if(ret <= 0)		//----------ret<=0时检查异常, 决定是否主动解除连接----------
                {
                    if(errno == EAGAIN || errno == EINTR || errno == EWOULDBLOCK)
                    ;
                    else if(ret==0)
                        {
                              //printf("服务器与客户端连接关闭 accept close : %d\r\n", events[j].data.fd);
                              // 向epoll删除client_sockfd监听事件
                            //ev.events = EPOLLIN|EPOLLET;
                            ev.data.fd = events[j].data.fd;
                            if(epoll_ctl(epoll_fd , EPOLL_CTL_DEL , events[j].data.fd , &ev) < 0)
                            {
                                    printf("server epoll_ctl : EPOLL_CTL_DEL failed !\r\n");
                                    if(shm!=NULL)        shmdt(shm);
                                    close(epoll_fd);
                                    exit(1);
                            }
                            removeRet=arrayRemoveItem(wss->client_fd_array, EPOLL_RESPOND_NUM, events[j].data.fd);    // 从数组剔除fd
                            int cfd=events[j].data.fd;
                            close(events[j].data.fd);	//关闭通道
                            if(wss->callCloseed!=NULL)
                            {
                            //wss->callCloseed(cfd);
                                pthreadArg *pArg=(pthreadArg*)malloc(sizeof(pthreadArg));
                                memset(pArg,0,sizeof(pthreadArg));
                                pArg->fd=cfd;
                                pArg->datatype=datatype;
                                pArg->buf=NULL;
                                pArg->bufLen=0;
                                pArg->callBack=NULL;
                                pArg->callCloseed=wss->callCloseed;
                                pthread_t thread_id;
                                if(pthread_create(&thread_id, NULL, thread_closeedfun, (void *)(pArg)) != 0)
                                    {
                                            printf("创建closeed处理线程失败!\r\n");
                                            if(pArg) free(pArg);
                                            continue;
                                    }
                            }
                            sprintf(fifo_buf.status,"closeed");

                    }
                    else{
                        printf("未知错误，读取句柄%d数据时发生,errno:%d\r\n", events[j].data.fd,errno);
                        continue;
                    }
                }


               
                }
			//===================发送数据事件===================
			else if(events[j].events & EPOLLOUT)
			    ;

            //向共享内存写入数据
            if(shm!=NULL)
            {
                if(thread_ipc>0){
                    //printf("IPC线程%d\n",thread_ipc);
                    pthread_cancel(thread_ipc);
                }                
                if(pthread_create(&thread_ipc, NULL, thread_ipc_fun, (void *)(&fifo_buf)) != 0)
                {
                    printf("创建IPC处理线程失败!\r\n");
                    continue;
                }
            }
		}
    }
s_end:
    webSocket_bufFree(&buf);
    struct msg_form msg;
    msg.mtype = 9753;
    msg.mtext = 'q';
    msgsnd(msqid, &msg, sizeof(msg.mtext), 0);
    // 断开连接
    if(shm!=NULL){
        shmdt(shm);
        /*删除共享内存、消息队列、信号量*/
        shmctl(shmid, IPC_RMID, &buf1);
        msgctl(msqid, IPC_RMID, &buf2);
        del_sem(semid);
    }
	//关闭epoll句柄
    close(epoll_fd);
    //关闭socket
    close(wss->fd);
}


/**********************************************
 * 服务器初始化
 * 参数：prot 端口
 *      server_callBack 服务回调
 *      server_callCloseed 连接关闭后事件
 * 返回值：
 * ***************************************/
int server_init(int port,CallBackFun *server_callBack,CallCloseedFun *server_callCloseed)
{
    //int exitFlag;
    //int i, client_fd;
    pthread_t sever_thread_id;
    WebSocket_Server wss;
    
    //===== 初始化服务器参数 =====
    memset(&wss, 0, sizeof(wss));
    //strcpy(wss.ip, "127.0.0.1");     
    wss.port = port;
    wss.callBack = server_callBack; // 响应客户端时, 需要干嘛?
    wss.callCloseed=server_callCloseed;
    //===== 开辟线程, 管理服务器 =====
    if(pthread_create(&sever_thread_id, NULL, server_thread_fun, (void *)(&wss)) != 0)
	{
		printf("创建服务失败!\r\n");
		return -1;
	} 
    pthread_join(sever_thread_id,NULL);
	//
   /* exitFlag = 0;
	while(!exitFlag)
	{
        // 每3秒推送信息给所有客户端

	    for(i = 0; i < EPOLL_RESPOND_NUM; i++)
	    {
	        if(wss.client_fd_array[i][1] != 0 && wss.client_fd_array[i][0] > 0)
	        {
	            client_fd = wss.client_fd_array[i][0];
	            if(webSocket_send(client_fd, "心跳数据", strlen("心跳数据"), false, WDT_TXTDATA) < 0)
                {
                    printf("server webSocket_send err !!\r\n");
                    exitFlag = 1;
                    break;
                }
	        }
            }
		exitFlag=1;
	    webSocket_delayms(3000);
        }*/
	
	//==============================
    //pthread_cancel(sever_thread_id);     // 等待线程关闭
    //printf("server close !\r\n");
	pthread_exit(0);
   // return 0;
}







